package hit.edu.zjc.Operator;


import hit.edu.zjc.Tool.ObjectConvert;
import hit.edu.zjc.Tool.RWObject;
import hit.edu.zjc.Tool.ShowOutPut;
import hit.edu.zjc.marshalling.ByteSerializerDeserializer;
import hit.edu.zjc.parsers.ByteTupleParserFactory;

import java.io.BufferedOutputStream;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.File;
import java.io.FileInputStream;
import java.io.FileNotFoundException;
import java.io.InputStream;
import java.io.ObjectInputStream;
import java.io.ObjectOutputStream;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
import java.util.Enumeration;
import java.util.HashSet;
import java.util.Hashtable;
import java.util.List;

import edu.uci.ics.hyracks.api.context.IHyracksTaskContext;
import edu.uci.ics.hyracks.api.dataflow.IOperatorNodePushable;
import edu.uci.ics.hyracks.api.dataflow.value.IRecordDescriptorProvider;
import edu.uci.ics.hyracks.api.dataflow.value.ISerializerDeserializer;
import edu.uci.ics.hyracks.api.dataflow.value.RecordDescriptor;
import edu.uci.ics.hyracks.api.exceptions.HyracksDataException;
import edu.uci.ics.hyracks.api.job.IOperatorDescriptorRegistry;
import edu.uci.ics.hyracks.dataflow.common.data.marshalling.UTF8StringSerializerDeserializer;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractSingleActivityOperatorDescriptor;
import edu.uci.ics.hyracks.dataflow.std.base.AbstractUnaryOutputSourceOperatorNodePushable;
import edu.uci.ics.hyracks.dataflow.std.file.IFileSplitProvider;
import edu.uci.ics.hyracks.dataflow.std.file.ITupleParser;

public class SourceGramMapIntegrationOperatorDescriptor extends AbstractSingleActivityOperatorDescriptor {
    private static final long serialVersionUID = 1L;
    private String[] FilePaths;
    
    public SourceGramMapIntegrationOperatorDescriptor(IOperatorDescriptorRegistry spec,String[] files) {
        super(spec, 0, 1);  
        recordDescriptors[0]  = new RecordDescriptor(new ISerializerDeserializer[]{ByteSerializerDeserializer.INSTANCE});
        FilePaths=files;
    }

    @Override
    public IOperatorNodePushable createPushRuntime(IHyracksTaskContext ctx,
            IRecordDescriptorProvider recordDescProvider, int partition, int nPartitions) {
    	 final ITupleParser parser =new ByteTupleParserFactory().createTupleParser(ctx);
    	return new AbstractUnaryOutputSourceOperatorNodePushable() {
        	
        	   /*
        	    *  将第二个GramMap，合并到第一GramMap上。
        	    * */
        	  public void MergeTwoGramMap(Hashtable<String,HashSet<Integer>>[] GramMap,Hashtable<String,HashSet<Integer>>[] OtherGramMap)
        	  {
        		  
        		  for(int i=0;i<GramMap.length;i++)
        		  {
        				Enumeration<String> e=OtherGramMap[i].keys();
        				while(e.hasMoreElements())
        				{
        					String key=e.nextElement();
        					HashSet<Integer> otherset=OtherGramMap[i].get(key);
        					
        					HashSet<Integer> set=GramMap[i].get(key);
        					if(set==null)//键不存在。
                			{
                			set=new HashSet<Integer>();
                			}
        					set.addAll(otherset);//并集。
        					GramMap[i].put(key, set);
        				}
        		  }
        	  }
        	
            @Override
            public void initialize() throws HyracksDataException {

            	final int filelength=FilePaths.length;
            	Hashtable<String,HashSet<Integer>>[] GramMap=(Hashtable<String,HashSet<Integer>>[])RWObject.ReadObject(FilePaths[0]);
            	//把其他GramMap合并到第一个上。
            	for(int i=1;i<filelength;i++)
            	{
            		Hashtable<String,HashSet<Integer>>[] OtherGramMap=(Hashtable<String,HashSet<Integer>>[])RWObject.ReadObject(FilePaths[i]);
            		MergeTwoGramMap(GramMap,OtherGramMap);
            	}
            	//合并结束。
      	
             writer.open();
             try
             	{
	     	    	byte[] GramMapBytes =ObjectConvert.ObjectToByteArray(GramMap);  		   
	     		   ByteArrayInputStream bais = new ByteArrayInputStream(GramMapBytes);	
	     	    	parser.parse(bais, writer);
	     	      bais.close();
	            
             	}
             catch(Exception e)
             	{
            	 	writer.fail();
            	 	e.printStackTrace();
             	}
             finally
             	{
            	 	writer.close();
             	}
            }
        };
    }
}